package com.yeming.flink.practice.transformation

import com.yeming.flink.practice.source.{MyCustomerSource, StationLog}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object TranSplit {

  //split已经被弃用了！
  def main(args: Array[String]): Unit = {

    //环境设置
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
//    streamEnv.setParallelism(1)
    //添加数据源
    val stream: DataStream[StationLog] = streamEnv.addSource(new MyCustomerSource)
    //数据转换
    val outputTag1 = new OutputTag[StationLog]("Success")
    val outputTag2 = new OutputTag[StationLog]("No Success")
    val outSideStream: DataStream[StationLog] = stream.process(new ProcessFunction[StationLog, StationLog] {
      override def processElement(value: StationLog, ctx: ProcessFunction[StationLog, StationLog]#Context, out: Collector[StationLog]): Unit = {
        out.collect(value)

        if (value.callType.equals("success")) {
          ctx.output(outputTag1, value)
        } else {
          ctx.output(outputTag2, value)
        }
      }
    })
    val stream1: DataStream[StationLog] = outSideStream.getSideOutput(outputTag1)
    val stream2: DataStream[StationLog] = outSideStream.getSideOutput(outputTag2)

    //数据sink
    stream1.print("通话成功！")
    stream2.print("通话失败！")
    outSideStream.print("所有数据！")
    //执行数据流
    streamEnv.execute("执行数据流转换")

  }

}
